#include "PFarm.h"
#include "Debug.h"


 PFarm::PFarm(DbParam dbParameters, const char *readers,const char *writers, bool debug)
: Farm(dbParameters)
{
	initilize(readers, writers);
	this->debug_ = debug;
}
PFarm:: ~PFarm()
{
	this->closeAll();
}

bool PFarm::initilize(const char *readers,const char *writers)
{
        char **hosts = g_strsplit(readers, HOST_HOST_SPLIT, -1);
        int hosts_num =  g_strv_length(hosts);
        for(int i=0; i< hosts_num ; i++ )
        {
                char **hps = g_strsplit(hosts[i], HOST_PORT_SPLIT, -1);
                int  hps_num =  g_strv_length(hps);
                PDbHost *db = NULL;
                if(hps_num == 1)
                {
                        db = new PDbHost();
                        db->dbHost_ = g_strstrip(hps[0]);
                        db->dbPort_ = myDb_.dbPort_;
                }
                else if(hps_num == 2)
                {
                        db = new PDbHost();
                        db->dbHost_ = g_strstrip(hps[0]);
                        db->dbPort_ = atoi(hps[1] );
                }
                else
                {
                   ERRO << hosts[i] <<  " is error! "<<endl;
                   abort();
		}
		if(db != NULL)
		{
			db->deadTime_ = 0;
			db->dbLive_ = true;
			MYSQL *c = NULL;
			c =mysql_init(NULL);
#ifdef MYSQL_OPT_RECONN    // both server and client at 5.0.16
			my_bool  my_true = true;
			mysql_options(c, MYSQL_OPT_RECONNECT, &my_true);
#endif
			DEBUG5(myDb_.dbUser_.c_str()," : ", myDb_.dbPass_.c_str()," : ", myDb_.dbName_.c_str());
                        mysql_options(c, MYSQL_SET_CHARSET_NAME,  myDb_.dbCharset_.c_str());

			if(NULL == mysql_real_connect( c,
						db->dbHost_.c_str(),
						myDb_.dbUser_.c_str(),
						myDb_.dbPass_.c_str(),
						myDb_.dbName_.c_str(),
						db->dbPort_, 
						NULL,
						0               /* my_bool compress */
						))
			{
				cerr <<"pconnect "<< db->dbHost_.c_str()<<" fail!"<<mysql_error(c) << endl;
				mysql_close(c);
				c = NULL;
				db->dbLive_ = false;
				db->deadTime_ = time(NULL);
			}
			db->pDb = c;
			readers_.push_back(db);
		}
                g_strfreev (hps);
        }
        g_strfreev (hosts );

        hosts = g_strsplit(writers, HOST_HOST_SPLIT, -1);
        hosts_num =  g_strv_length(hosts);
        for(int i=0; i< hosts_num ; i++ )
        {
                char **hps = g_strsplit(hosts[i], HOST_PORT_SPLIT, -1);
                int  hps_num =  g_strv_length(hps);
                PDbHost *db = NULL;
                if(hps_num == 1)
                {
                        db = new PDbHost();
                        db->dbHost_ =g_strstrip(hps[0]);
                        db->dbPort_ = myDb_.dbPort_;
                }
                else if(hps_num == 2)
                {
                        db = new PDbHost();
                        db->dbHost_ = g_strstrip(hps[0]);
                        db->dbPort_ = atoi(hps[1] );
                }
                else
                {
                   ERRO << hosts[i] <<  " is error! "<<endl;
                   abort();
		}
		if(db != NULL)
		{
			db->deadTime_ = 0;
			db->dbLive_ = true;
			MYSQL *c = NULL;
			c =mysql_init(NULL);
#ifdef MYSQL_OPT_RECONN    // both server and client at 5.0.16
			my_bool  my_true = true;
			mysql_options(c, MYSQL_OPT_RECONNECT, &my_true);
#endif
			DEBUG5(myDb_.dbUser_.c_str()," : ", myDb_.dbPass_.c_str()," : ", myDb_.dbName_.c_str());
                        mysql_options(c, MYSQL_SET_CHARSET_NAME,  myDb_.dbCharset_.c_str());

			if(NULL == mysql_real_connect( c,
						db->dbHost_.c_str(),
						myDb_.dbUser_.c_str(),
						myDb_.dbPass_.c_str(),
						myDb_.dbName_.c_str(),
						db->dbPort_, 
						NULL,
						0               /* my_bool compress */
						))
			{
                                ERRO << mysql_error(c) << endl;
                                mysql_close(c);
                                c = NULL;
				db->dbLive_ = false;
				db->deadTime_ = time(NULL);
			}
			db->pDb = c;
			writers_.push_back(db);
		}
                g_strfreev (hps);
        }
        g_strfreev (hosts );
        rCount_ = 0;
        return true;
}

MYSQL *PFarm:: GetWriter(int idx)
{
#ifdef _THREAD_SAFE
	DEBUG1("+++++++++++++++++++++++   pthread_mutex_lock, please sure to use unlock");
	pthread_mutex_lock(&_lock);
#endif
	int size = (int)writers_.size();
	if( idx<0 || idx >= size )
	{
		error_ = "writer's idx is out bounder";
		return NULL;
	}

	PDbHost *dh = (PDbHost *) writers_[idx];
	if(dh->pDb == NULL)
	{
		dh->pDb =mysql_init(NULL);
#ifdef MYSQL_OPT_RECONN    // both server and client at 5.0.16
		my_bool  my_true = true;
		mysql_options(dh->pDb, MYSQL_OPT_RECONNECT, &my_true);
#endif
                mysql_options(dh->pDb, MYSQL_SET_CHARSET_NAME,  myDb_.dbCharset_.c_str());

		if(NULL == mysql_real_connect(dh->pDb,
					dh->dbHost_.c_str(),
					myDb_.dbUser_.c_str(),
					myDb_.dbPass_.c_str(),
					myDb_.dbName_.c_str(),
					dh->dbPort_, 
					NULL,
					0               /* my_bool compress */
					))
		{
                        ERRO << mysql_error(dh->pDb) << endl;
                        mysql_close(dh->pDb);
                        dh->pDb = NULL;
			dh->dbLive_ = false;
			dh->deadTime_ = time(NULL);
		}
                return dh->pDb;
	}
	if(0 == mysql_ping( dh->pDb ))
		return dh->pDb;
	else
		ERRO << mysql_error(dh->pDb) << endl;
	if ( 0 == mysql_ping( dh->pDb))
		return dh->pDb;
	else	   
		ERRO << mysql_error(dh->pDb) << endl;
	return NULL;
}

MYSQL* PFarm::GetReader()
{
#ifdef   _THREAD_SAFE
     DEBUG1("+++++++++++++++++++++++     pthread_mutex_lock, please sure to unlock");
     pthread_mutex_lock(&_lock);
#endif
     int size = (int)readers_.size();
     if(size <= 0)
             return NULL;
     int num = 0;
     MYSQL *c = NULL;
     while(num < size )
     {
	     c = getNextReader(num, size);
	     if(c != NULL)
	     {
		     DEBUG2("get reader conn, num =", num);
		     return c;
	     }
     }
     error_ = "can't connection all readers";
     return NULL;
}

MYSQL* PFarm::getNextReader(int &num, int size )
{
	num++;
	rCount_ = rCount_%size;
	PDbHost *dh = (PDbHost *) readers_[ rCount_ ++ ];
	if(!dh->dbLive_)
	{
		long lt = time(NULL);
		if(dh->deadTime_ > lt - 300)
		{
			DEBUG4("db is down but time not timeout, so return NULL : deadTime=",dh->deadTime_, "; now=", lt);
			return NULL;
		}
		dh->dbLive_ = true;
	}

	if(dh->pDb == NULL)
	{
		dh->pDb =mysql_init(NULL);
#ifdef MYSQL_OPT_RECONN    // both server and client at 5.0.16
		my_bool  my_true = true;
		mysql_options(dh->pDb, MYSQL_OPT_RECONNECT, &my_true);
#endif
                mysql_options(dh->pDb, MYSQL_SET_CHARSET_NAME,  myDb_.dbCharset_.c_str());

		if(NULL == mysql_real_connect(dh->pDb,
					dh->dbHost_.c_str(),
					myDb_.dbUser_.c_str(),
					myDb_.dbPass_.c_str(),
					myDb_.dbName_.c_str(),
					dh->dbPort_, 
					NULL,
					0               /* my_bool compress */
					))
		{
                        ERRO << mysql_error(dh->pDb) << endl;
                        mysql_close(dh->pDb);
                        dh->pDb = NULL;
			dh->dbLive_ = false;
			dh->deadTime_ = time(NULL);
		}
                return dh->pDb;
	}
        DEBUG1("ping to get handle");
	if(0 == mysql_ping( dh->pDb ))
         	return dh->pDb;
	else
		ERRO << mysql_error(dh->pDb) << endl;
	if ( 0 == mysql_ping( dh->pDb))
		return dh->pDb;
	else	   
	{
		ERRO << mysql_error(dh->pDb) << endl;
		dh->dbLive_ = false;
		return NULL;
	}
}

void PFarm::closeAll()
{
#ifdef _THREAD_SAFE
    pthread_mutex_lock(&_lock);
    DEBUG1("++++++++++++++++++++++      pthread_mutex_lock! please sure to unlock!");
#endif
 
	ERRO<<"Do you sure that it's the end of program life? else you are wrong!"<<endl;
	for (size_t i=0; i < readers_.size(); ++i)
	{      
		PDbHost *dh = (PDbHost *) readers_[ (int)i ];
		if(dh->pDb != NULL)
		{
			DEBUG1("close reader");
			mysql_close( dh->pDb);
			dh->pDb = NULL;
		}
	}
	for (size_t i=0; i < writers_.size(); ++i)
	{      
		PDbHost *dh = (PDbHost *) writers_[ (int)i ];
		if(dh->pDb != NULL)
		{
			DEBUG1("close writer");
			mysql_close( dh->pDb);
			dh->pDb = NULL;
		}
	}
#ifdef _THREAD_SAFE
    DEBUG1("-----------------------         pthread_mutex_unlock!");
    pthread_mutex_unlock(&_lock);
#endif
}

void PFarm::CloseConnection(MYSQL *c, bool write) 
{
#ifdef _THREAD_SAFE
    DEBUG1("-----------------------        pthread_mutex_unlock!");
    pthread_mutex_unlock(&_lock);
#endif
    DEBUG1("all is persistant, can't be close!");
} 

void PFarm:: Print()
{
#ifdef _THREAD_SAFE
        DEBUG1("++++++++++++++++++++++++    pthread_mutex_lock, please sure to use unlock");
        pthread_mutex_lock(&_lock);
#endif

        cout<<"PFarm writers ::"<<endl;
        for (size_t i=0; i < writers_.size(); ++i)
        {      
                PDbHost *dh = (PDbHost *) writers_[ (int)i ];
                cout<< "w["<< i<<"] = "<< dh->dbHost_.c_str()<< ":" << dh->dbPort_ <<":"<< dh->deadTime_ <<":"<< dh->dbLive_<<endl;

        }

        cout<<"PFarm readers ::"<<endl;
        for (size_t i=0; i < readers_.size(); ++i)
        {      
                PDbHost *dh = (PDbHost *) readers_[ (int)i ];
                cout<< "r["<< i<<"] = "<< dh->dbHost_.c_str()<< ":" << dh->dbPort_ <<":"<< dh->deadTime_ <<":"<< dh->dbLive_<<endl;

        }
#ifdef _THREAD_SAFE
        DEBUG1("---------------------         pthread_mutex_unlock");
        pthread_mutex_unlock(&_lock);
#endif

}

